package longestExp;

import Bean.JobExperienceBean;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 * @author Isabella
 */
public class LongestExpReducer extends Reducer<Text, IntWritable, Text, IntWritable> {

    private final static IntWritable result = new IntWritable();
    private List<JobExperienceBean> jobList = new ArrayList<>();

    @Override
    public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        int maxExp = 0;

        for (IntWritable v : values) {
            int currentExp = v.get();
            maxExp = Math.max(maxExp, currentExp);
        }

        jobList.add(new JobExperienceBean(key.toString(), maxExp));
    }

    @Override
    protected void cleanup(Context context) throws IOException, InterruptedException {
        jobList.sort(null);

        int count = 0;
        for (JobExperienceBean job : jobList) {
            if (count < 10) {
                result.set(job.getExperience());
                context.write(new Text(job.getJob()), result);
                count++;
            } else {
                break;
            }
        }
    }
}
